package org.orbis.communicate.indexer.content;

import java.util.Map.Entry;
import java.util.Comparator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.PriorityBlockingQueue;
import org.msgpack.object.ArrayType;
import org.orbis.communicate.Configuration;
import org.orbis.communicate.hash.Hashing;
import org.orbis.communicate.lock.IndexableSwitch;
import org.orbis.communicate.node.Node;
import org.orbis.communicate.rpc.Communicator;
import org.orbis.index.IndexingPackage;
import org.orbis.logging.LogParser;

public class ContentIndexerHost implements Runnable{
	private static int replication;
	private static Configuration conf;
	private static PriorityBlockingQueue<IndexingPackage> hostQueue;
	private static ConcurrentHashMap<Node, IndexingPackage> suspendQueue;
	private static Communicator communicator;
	private int retryCount;

	static{
		hostQueue = new PriorityBlockingQueue<IndexingPackage>(10, new Comparator<IndexingPackage>(){
			@Override
			public int compare(IndexingPackage o1, IndexingPackage o2){
				if(o1 == null || o2 == null) return 0;
				if(o1.ID() > o2.ID()) return 1;
				if(o1.ID() < o2.ID()) return -1;
				return 0;
			}
		});
		suspendQueue = new ConcurrentHashMap<Node, IndexingPackage>();
	}

	public ContentIndexerHost(){}

	public static void property(int repl, Communicator comm, Configuration config){
		replication = repl;
		communicator = comm;
		conf = config;
	}

	// FIXME: Index in Indexing package
	public static void addQueue(IndexingPackage newIndex){
		hostQueue.offer(newIndex);
	}

	public static boolean isEmpty(){
		return hostQueue.isEmpty();
	}

	public static int residue(){
		return hostQueue.size();
	}

	public static void resumeQueue(Node liveNode){
		for(Entry<Node, IndexingPackage> entries: suspendQueue.entrySet()){
			if(entries.getKey() == liveNode){
				hostQueue.add(entries.getValue());
				suspendQueue.remove(entries);
			}
		}
	}

	// Drop from suspend queue
	public static void nodeDead(Node deadNode){
		for(Entry<Node, IndexingPackage> entries: suspendQueue.entrySet()){
			if(entries.getKey() == deadNode) suspendQueue.remove(entries);
		}
	}

	@Override
	public void run(){
		retryCount = 2;
		long startTime = 0;
		long hash;
		Entry<Long, Node> currentEntry = null;
		Node myNode = null;
		IndexingPackage newIndex = null;

		while(true){
			try{
				newIndex = hostQueue.take();
			}catch(InterruptedException e1){
				continue;
			}
			if(myNode == null) myNode = conf.myNode();
			hash = Hashing.getContentHash(newIndex.ID());
			currentEntry = conf.ceilingHashEntry(hash);
			
			Node nearestNode = currentEntry.getValue();
			int cluster = conf.getNodeIndex(nearestNode);
			newIndex.setCluster(cluster);

			startTime = LogParser.initTime();

			LogParser.timerFinest("Index / calc node", startTime);
			ArrayType contentObjs = null;

			// Stop to post to the client queues, due to keep replication in
			// time
			try{
				IndexableSwitch.waitIfLocked();
			}catch(InterruptedException e){}
			int nodeSize = conf.nodeList().size();
			for(int i = 0; i < (replication > nodeSize? nodeSize: replication); i++){
				if(i > 0) currentEntry = conf.higherHashEntry(currentEntry.getKey());

				LogParser.finest("Store content to " + currentEntry.getValue().getAddr());
				if(currentEntry.getValue().equals(myNode)){
					// local: Directly adds to client queue
					ContentIndexerClient.addToQueue(newIndex);
				}else if(currentEntry.getValue().isSuspend()){
					suspendQueue.put(currentEntry.getValue(), newIndex);
				}else{
					// remote
					startTime = LogParser.initTime();
					if(contentObjs == null) contentObjs = newIndex.toPackageContent();
					for(int j = 0; j < retryCount; j++){
						try{
							communicator.indexContentRemote(currentEntry.getValue(), contentObjs);
							break;
						}catch(Exception e){
							e.printStackTrace();
							LogParser.warning(String.format("Can not index %d due to server down.",
								newIndex.ID()));
							LogParser.warning("Retry to index in next node.");
						}
					}
					LogParser.timerFinest("Index / Posting to remote", startTime);
				}
			}
		}
	}
}
